package com.oracle.calllog.etl.mapper;

import java.io.IOException;

import com.oracle.calllog.etl.util.Constants;
import com.oracle.calllog.etl.util.HBaseUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.oracle.calllog.etl.model.Calllog;

/******************************************
 * @ClassName CalllogsEtl2HBaseMapper
 * @Description ETL流程的Mapper类
 * @Author Mr.Chen
 * @Date 2022/6/11 21:13
 * @Version 1.0
 *****************************************/
public class CalllogsEtl2HBaseMapper extends Mapper<LongWritable, Text, Text, Put>{
    //用于记录RegionCode的加盐信息的属性，在输出到下一个环节时可以按分区编码聚合数据，减少IO次数提升HBase写数据性能
    private Text regionCodeKey = new Text();

    //用于校验手机号码的正则表达式,本属性不参与输出到下一个环节
    private String mobileRegExp = "1((3[0-9])|(4[14-8])|(5[0-35-9])|(6[5-7])|(7[014-8])|(8[0-9])|(9[189]))\\d{8}";


    /**
     * map方法用于完成ETL主体业务逻辑实现，并将向HBase中存储的数据格式组织为Put类型的对象
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        // 1 获取1行
        String line = value.toString();

        // 2 解析日志是否合法，并转换为主叫人做为第1个联系人的bean对象
        Calllog contact1Bean = pressLog(line);

        // 3 清洗垃圾数据
        if (!contact1Bean.isValid()) {
            //设置无效日志的计数器
            context.getCounter("CallLogsCounter", "Invalid Data").increment(1);
            return;
        }
        // 设置有效日志的计数器
        context.getCounter("CallLogsCounter", "Valid Data").increment(1);

        // 4 根据通话记录的Bean对象得到另一个将被叫人信息置为contact1的Bean对象，并将呼叫类型设置为[1:被叫人为contact1]
        Calllog contact2Bean = this.getContact2ConvertContact1Bean(contact1Bean);

        // 5 将Bean对象整理为用于分析通话时长和通话次数的Put对象数据
        String regionCode = null;
        Put put = null;
        // 5.1 生成主叫人做为第1联系人的通话数据所需要加的盐(RegionCode)，以此确定数据应存入哪个分区；
        //      并将RegionCode做为输出到下一个环节的Key，完成以RegionCode为依据的Put对象数据归并
        regionCode = HBaseUtil.genRegionCode(contact1Bean.getContact1No(),contact1Bean.getBuildTime(), Constants.REGION_NUMS);
        put = this.generatePutByBean(regionCode, contact1Bean);
        this.regionCodeKey.set(regionCode);
        // 将数据K-V输出到下一个环节
        context.write(this.regionCodeKey, put);

        // 5.2 生成被叫人做为第1联系人的通话数据所需要加的盐(RegionCode)，以此确定数据应存入哪个分区；
        //      并将RegionCode做为输出到下一个环节的Key，完成以RegionCode为依据的Put对象数据归并
        regionCode = HBaseUtil.genRegionCode(contact2Bean.getContact1No(),contact2Bean.getBuildTime(), Constants.REGION_NUMS);
        put = this.generatePutByBean(regionCode, contact2Bean);
        this.regionCodeKey.set(regionCode);
        // 将数据K-V输出到下一个环节
        context.write(this.regionCodeKey, put);
    }


    /**
     * 解析一条日志文本，并将无效数据状态置为false的方法
     * @param line 一条通话日志的文本
     * @return 一条通话日志信息的Bean对象
     */
    private Calllog pressLog(String line) {
        Calllog calllog = new Calllog();

        // 截取
        String[] fields = line.split(" ");

        //过滤数据不完整记录(没有接通状态与通话时长信息的通话日志 && 基站通信线程同步异常产生的数据都为NULL信息的通话日志)
        if (fields.length == 14 && line.indexOf("NULL")==-1) {
            // 2封装数据
            calllog.setCallId(fields[0]);
            calllog.setContact1Division(fields[1]);
            calllog.setContact1NetType(fields[2]);
            calllog.setContact1BizType(fields[3]);
            calllog.setContact1No(fields[4]);
            calllog.setContact1Name(fields[5]);
            calllog.setContact2Division(fields[6]);
            calllog.setContact2NetType(fields[7]);
            calllog.setContact2OperatorType(fields[8]);
            calllog.setContact2No(fields[9]);
            calllog.setContact2Name(fields[10]);
            calllog.setBuildTime(fields[11].replaceAll("[\\D]+", ""));
            calllog.setCallDuration(fields[12]);
            calllog.setConnectType(fields[13]);

            //设置为主叫人为第1个联系人的标识
            calllog.setCallFlag("0");
            //设置为有效通话日志标识
            calllog.setValid(true);

            //如果被叫人手机号码为空(无被叫人信息)或被叫人手机号不是常规运营商的手机号码,则视为垃圾数据
            if(calllog.getContact2No()==null || "".equals(calllog.getContact2No())
                    || "*".equals(calllog.getContact2OperatorType()) || !calllog.getContact2No().matches(mobileRegExp)) {
                //设置为无效通话日志标识
                calllog.setValid(false);
            }

//			// 未接通的通话记录在有些通话记录指标统计中为垃圾数据，标注为无效数据
//			if ("LOSS_CALL".equals(calllog.getConnectType())) {
//				calllog.setValid(false);
//			}
        }else {
            calllog.setValid(false);
        }

        return calllog;
    }


    /**
     * 获取一个将【主叫人做为第1联系人,被叫人做为第2联系人】的Bean转换为【被叫人做为第1联系人,主叫人做为第2联系人】的Bean <BR/>
     *  冗余一条记录以提升统计计算的性能
     * @param contact1Bean 【主叫人做为第1联系人,被叫人做为第2联系人】的Bean
     * @return 【被叫人做为第1联系人,主叫人做为第2联系人】的Bean
     */
    private Calllog getContact2ConvertContact1Bean(Calllog contact1Bean){
        Calllog contact2Bean = new Calllog();
        contact2Bean.setContact1No(contact1Bean.getContact2No());
        contact2Bean.setContact1Name(contact1Bean.getContact2Name());
        contact2Bean.setContact1Division(contact1Bean.getContact2Division());
        contact2Bean.setContact1NetType(contact1Bean.getContact2NetType());
        contact2Bean.setContact1BizType(contact1Bean.getContact2OperatorType());

        contact2Bean.setContact2No(contact1Bean.getContact1No());
        contact2Bean.setContact2Name(contact1Bean.getContact1Name());
        contact2Bean.setContact2Division(contact1Bean.getContact1Division());
        contact2Bean.setContact2NetType(contact1Bean.getContact1NetType());
        contact2Bean.setContact2OperatorType(contact1Bean.getContact1BizType());

        contact2Bean.setCallId(contact1Bean.getCallId());
        contact2Bean.setBuildTime(contact1Bean.getBuildTime());
        contact2Bean.setCallDuration(contact1Bean.getCallDuration());
        contact2Bean.setConnectType(contact1Bean.getConnectType());

        contact2Bean.setCallFlag("1");
        contact2Bean.setValid(true);

        return contact2Bean;
    }


    /**
     * 根据Bean对象，生成对应的Rowkey并组织成一个用于向HBase中新增或修改的Put描述对象
     * @param bean 一条通话日志的bean
     * @return 一个put对象
     * @see HBaseUtil#genRegionCode(String,String,int)
     * @see HBaseUtil#genRowKey(String,String,String,String,String,String)
     */
    private Put generatePutByBean(String regionCode, Calllog bean){

        // 根据主要数据参考，生成在HBase中存储的Rowkey
        String rowKey = HBaseUtil.genRowKey(regionCode, bean.getContact1No(),bean.getBuildTime(), bean.getContact2No(),bean.getCallDuration(),bean.getCallFlag());
        // 根据rowKey创建向HBase中新增或修改数据的描述对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 组织put对象
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT1_NO), Bytes.toBytes(bean.getContact1No()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT1_NAME), Bytes.toBytes(bean.getContact1Name()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT2_NO), Bytes.toBytes(bean.getContact2No()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT2_NAME), Bytes.toBytes(bean.getContact2Name()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_BUILD_TIME), Bytes.toBytes(bean.getBuildTime()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_DURATION), Bytes.toBytes(bean.getCallDuration()));
        //根据业务规则和集群性能，以下属性也可以不处理
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CALL_ID), Bytes.toBytes(bean.getCallId()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT1_DIVISION), Bytes.toBytes(bean.getContact1Division()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT1_NET_TYPE), Bytes.toBytes(bean.getContact1NetType()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT1_BIZ_TYPE), Bytes.toBytes(bean.getContact1BizType()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT2_DIVISION), Bytes.toBytes(bean.getContact2Division()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT2_OPERATOR_TYPE), Bytes.toBytes(bean.getContact2OperatorType()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT2_NET_TYPE), Bytes.toBytes(bean.getContact2NetType()));
        put.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONNECT_TYPE), Bytes.toBytes(bean.getConnectType()));

        return put;
    }
}

